package org.budo.dubbo.protocol.async.repository.rocketmq;

import java.util.List;
import java.util.Set;

import org.apache.rocketmq.client.consumer.MQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.budo.dubbo.protocol.async.repository.BudoAsyncInvocation;
import org.budo.dubbo.protocol.async.util.AsyncSerializeUtil;
import org.budo.dubbo.protocol.async.util.DestinationUtil;

import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcInvocation;

/**
 * @author limingwei
 * 
 * @see http://rocketmq.apache.org/docs/simple-example/
 */
public class RocketMQAsyncRepository extends AbstractRocketMQAsyncRepository {
    @Override
    public void sendMessageWhenInvoke(Invocation invocation) {
        try {
            String destinationName = DestinationUtil.getDestinationName(invocation.getInvoker());

            BudoAsyncInvocation budoAsyncInvocation = BudoAsyncInvocation.fromDubboInvocation(invocation);
            byte[] requestBody = AsyncSerializeUtil.serializeAsyncInvocation(budoAsyncInvocation, invocation.getInvoker());

            Message message = new Message(destinationName, requestBody);

            this.getRocketMQProducer().send(message);
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Integer popMessageAndInvoke(final Invoker invoker) {
        try {
            String destinationName = DestinationUtil.getDestinationName(invoker);

            MQPullConsumer mqPullConsumer = this.getRocketMQPullConsumer();

            Set<MessageQueue> messageQueues = mqPullConsumer.fetchSubscribeMessageQueues(destinationName);

            int messageCount = 0;

            for (MessageQueue messageQueue : messageQueues) {
                String subExpression = null;
                long offset = 0;
                int maxNums = 0;
                PullResult pullResult = mqPullConsumer.pullBlockIfNotFound(messageQueue, subExpression, offset, maxNums);

                List<MessageExt> messages = pullResult.getMsgFoundList();
                for (MessageExt message : messages) {
                    byte[] messageBody = message.getBody();
                    BudoAsyncInvocation budoAsyncInvocation = AsyncSerializeUtil.deserializeAsyncInvocation(messageBody, invoker);

                    RpcInvocation invocation = BudoAsyncInvocation.toDubboInvocation(budoAsyncInvocation);
                    invoker.invoke(invocation); // 执行方法

                    messageCount++;
                }
            }

            return messageCount;
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void sendMessage(String destinationName, Object message) {
        throw new RuntimeException();
    }

    @Override
    public void sendRawMessage(String destinationName, Object rawMessage) {
        throw new RuntimeException();
    }
}